package com.shujichen.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import com.shujichen.entity.CompileTask;
import com.shujichen.enums.TaskStatusEnum;
import com.shujichen.service.TaskService;
import com.shujichen.utils.DateUtil;
import com.shujichen.utils.PoetryCacheRedis;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

@Slf4j
@Service
public class TaskServiceImpl implements TaskService {

    @Autowired
    private ExecutorService taskExecutor;

    // 静态计数器
    private static int counter = 1;
    // 任务缓存前缀
    public static final String TASK_PREFIX = "task:";
    // 日期格式
    private static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("yyyyMMddHHmmss");

    // 等待队列及辅助数据结构
    private final BlockingQueue<CompileTask> taskQueue = new LinkedBlockingQueue<>();
    private final List<CompileTask> waitList = new CopyOnWriteArrayList<>();
    private final Map<String, Integer> taskIndexMap = new ConcurrentHashMap<>();

    // 运行中任务跟踪
    private final Map<String, Future<?>> runningTasks = new ConcurrentHashMap<>();

    // 监控指标
    private final AtomicInteger maxQueueSize = new AtomicInteger(0);
    private final AtomicLong totalProcessed = new AtomicLong(0);

    // 初始化执行器
    @PostConstruct
    public void init() {
        startTaskConsumer();
        startTimeoutMonitor();
        startQueueMonitor();
    }

    // 优雅关闭
    @PreDestroy
    public void shutdown() {
        taskExecutor.shutdownNow();
        log.info("Task system shutdown, total processed tasks: {}", totalProcessed.get());
    }

    @Override
    public void add() {
        String taskId = UUID.randomUUID().toString();
        CompileTask task = new CompileTask();
        task.setTaskId(taskId);
        task.setTaskName(generate());
        task.setAddTime(new Date());
        task.setTaskStatus(TaskStatusEnum.WAITING.getDesc());

        try {
            // 原子化入队操作
            synchronized (this) {
                taskQueue.put(task);
                waitList.add(task);
                taskIndexMap.put(taskId, waitList.size() - 1);
                updateMaxQueueSize();
            }

            // 保存到Redis
            PoetryCacheRedis.put(TASK_PREFIX + task.getTaskId(), task);

            // 提交任务到执行器
            runningTasks.put(taskId, taskExecutor.submit(createTaskWrapper(task)));
            log.info("Task added: {}", taskId);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("Task add interrupted: {}", taskId);
        }
    }

    /**
     * 创建任务包装器（核心逻辑）
     *
     * @param task 任务实体
     * @return 可执行的任务包装
     */
    private Runnable createTaskWrapper(CompileTask task) {
        return () -> {
            try {
                // 前置状态更新
                updateTaskStatus(task, TaskStatusEnum.RUNNING);

                // 执行实际任务逻辑（示例）
                log.info("Executing task: {}", task.getTaskId());

                // 产生随机数, 如果大于超时时间则有几率产生超时任务
                int randomInt = new Random().nextInt(15);
                log.info("模拟任务执行随机数为 {}", randomInt);

                // 模拟业务执行异常
                if (Arrays.asList(3, 6, 9).contains(randomInt)) {
                    int num = 1 / 0;
                }

                // 模拟业务执行
                Thread.sleep(randomInt * 1000);

                // 成功状态更新
                updateTaskStatus(task, TaskStatusEnum.SUCCESS);
                log.info("SUCCESS task: {}", task.getTaskId());
            } catch (InterruptedException e) {
                updateTaskStatus(task, TaskStatusEnum.TIMEOUT);
                log.info("TIMEOUT task: {}", task.getTaskId());
            } catch (Exception e) {
                updateTaskFailedInfo(task, e);
                log.info("FAILED task: {}", task.getTaskId());
            } finally {
                // 后置清理
                cleanupTask(task);
                log.info("CleanupTask task: {}", task.getTaskId());
            }
        };
    }

    // 启动任务消费者
    private void startTaskConsumer() {
        new Thread(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    CompileTask task = taskQueue.poll(1, TimeUnit.SECONDS);
                    if (task != null) {
                        processTask(task);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }, "Task-Consumer-Thread").start();
    }

    // 处理单个任务
    private void processTask(CompileTask task) {
        try {
            Future<?> future = runningTasks.get(task.getTaskId());
            if (future != null) {
                future.get(task.getTimeout(), TimeUnit.SECONDS);
            }
        } catch (TimeoutException e) {
            updateTaskStatus(task, TaskStatusEnum.TIMEOUT);
        } catch (Exception e) {
            log.error("Task processing error: {}", task.getTaskId(), e);
        } finally {
            totalProcessed.incrementAndGet();
        }
    }

    // 超时监控
    private void startTimeoutMonitor() {
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(() -> {
            runningTasks.forEach((taskId, future) -> {
                CompileTask task = getCachedTask(taskId, CompileTask.class);
                if (task != null && task.getTaskStatus().equals(TaskStatusEnum.RUNNING.getDesc())) {
                    int timeout = task.getTimeout() * 1000;
                    long consume = System.currentTimeMillis() - task.getStartTime().getTime();
                    log.info("Task {}, 超时时间 {} ms, 执行时间 {} ms, 判定结果 {}", taskId, timeout, consume, consume > timeout);

                    if (consume > timeout) {
                        future.cancel(true);
                        runningTasks.remove(taskId);
                        updateTaskStatus(task, TaskStatusEnum.TIMEOUT);
                        log.warn("Task timeout: {}", taskId);
                    }
                }
            });
        }, 1, 1, TimeUnit.SECONDS);
    }

    // 队列监控（记录指标）
    private void startQueueMonitor() {
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
            System.out.println("==========任务队列监控状态==========");
            ThreadPoolExecutor executor = (ThreadPoolExecutor) taskExecutor;
            System.out.println("等待任务数: " + taskQueue.size());
            System.out.println("运行任务数: " + runningTasks.size());
            System.out.println("运行线程数: " + executor.getActiveCount());
            System.out.println("最大队列长度: " + maxQueueSize.get());
            System.out.println("总处理任务数: " + totalProcessed.get());
            System.out.println("==================================");
        }, 1, 5, TimeUnit.SECONDS);
    }

    // 获取等待位置（核心优化）
    private String getSort(String taskId) {
        Integer index = taskIndexMap.get(taskId);
        if (index != null) {
            // 返回前面等待的任务数（当前索引即为前面任务数）
            return String.valueOf(index);
        }
        // 任务可能已开始执行或不存在
        CompileTask task = getCachedTask(taskId, CompileTask.class);
        return (task != null && task.getTaskStatus().equals(TaskStatusEnum.WAITING.getDesc()))
                ? "0"
                : "-1";
    }

    // 更新最大队列长度
    private synchronized void updateMaxQueueSize() {
        int currentSize = waitList.size();
        if (currentSize > maxQueueSize.get()) {
            maxQueueSize.set(currentSize);
        }
    }

    // 统一状态更新方法
    private void updateTaskStatus(CompileTask task, TaskStatusEnum status) {
        task.setTaskStatus(status.getDesc());
        if (status == TaskStatusEnum.RUNNING) {
            task.setStartTime(new Date());
        } else {
            task.setEndTime(new Date());
        }
        PoetryCacheRedis.put(TASK_PREFIX + task.getTaskId(), task);
    }

    // 任务失败更新方法
    private void updateTaskFailedInfo(CompileTask task, Exception e) {
        task.setEndTime(new Date());
        task.setFailedInfo(e.getMessage());
        task.setTaskStatus(TaskStatusEnum.FAILED.getDesc());
        PoetryCacheRedis.put(TASK_PREFIX + task.getTaskId(), task);
    }

    // 任务清理
    private synchronized void cleanupTask(CompileTask task) {
        try {
            runningTasks.remove(task.getTaskId());
            waitList.remove(task);
            taskIndexMap.remove(task.getTaskId());
            // 更新后续任务的索引
            updateFollowingTaskIndexes();
        } catch (Exception e) {
            log.error("Task cleanup error: {}", task.getTaskId(), e);
        }
    }

    // 更新后续任务索引
    private void updateFollowingTaskIndexes() {
        synchronized (this) {
            for (int i = 0; i < waitList.size(); i++) {
                CompileTask t = waitList.get(i);
                taskIndexMap.put(t.getTaskId(), i);
            }
        }
    }

    // 其他方法保持原有结构，补充优化逻辑...

    @Override
    public List<CompileTask> query() {
        List<CompileTask> taskList = PoetryCacheRedis.findByPrefix(TASK_PREFIX, CompileTask.class);

        if (CollectionUtil.isNotEmpty(taskList)) {
            return taskList.stream()
                    .peek(this::calculateDurations)
                    .peek(task -> {
                        if (TaskStatusEnum.WAITING.getDesc().equals(task.getTaskStatus())) {
                            task.setTaskAwaitCount(getSort(task.getTaskId()));
                        } else {
                            task.setTaskAwaitCount("-");
                        }
                    })
                    .collect(Collectors.toList());
        }
        return Collections.emptyList();
    }

    // 计算耗时（抽取为独立方法）
    private void calculateDurations(CompileTask task) {
        Date addTime = task.getAddTime();
        Date startTime = task.getStartTime();
        Date endTime = task.getEndTime();

        if (addTime != null && startTime != null) {
            task.setTaskAwaitDuration(DateUtil.formatDuration(startTime.getTime() - addTime.getTime()));
        }

        if (startTime != null && endTime != null) {
            task.setTaskRealDuration(DateUtil.formatDuration(endTime.getTime() - startTime.getTime()));
        }

        if (addTime != null && endTime != null) {
            task.setPredictionDuration(DateUtil.formatDuration(endTime.getTime() - addTime.getTime()));
        }
    }

    @Override
    public void clear() {
        Collection<Object> keys = PoetryCacheRedis.keys();
        if (CollectionUtil.isNotEmpty(keys)) {
            keys.stream().forEach(key -> PoetryCacheRedis.remove(key.toString()));
        }
        reset();
    }

    private <T> T getCachedTask(String taskId, Class<T> clazz) {
        Object value = PoetryCacheRedis.get(TASK_PREFIX + taskId);
        if (value == null) {
            return null;
        }
        if (!clazz.isInstance(value)) {
            throw new IllegalArgumentException(
                    String.format("Value for key %s is not of type %s",
                            TASK_PREFIX + taskId, clazz.getName()));
        }
        return clazz.cast(value);
    }

    /**
     * 生成任务名称（格式：TASK-时间-序号）
     */
    public static synchronized String generate() {
        return String.format("TASK-%s-%03d",
                DATE_FORMAT.format(new Date()),
                counter++);
    }

    /**
     * 重置计数器（清空后调用）
     */
    public static synchronized void reset() {
        counter = 1;
        System.out.println("任务计数器已重置");
    }

}
